package com.neusoft.reactor.jdkflow;

import lombok.Data;

import java.util.concurrent.Flow;

/**
 * @author lichunlong lcl253137@alibaba-inc.com
 * @desc Reactive JDK框架"订阅者"默认实现
 * @date 2023-01-17
 */
@Data
public class StringSubscriber implements Flow.Subscriber<String> {

    /**
     * 订阅者名称
     */
    private final String subscriberName;

    private Flow.Subscription subscription;

    public StringSubscriber(String subscriberName) {
        this.subscriberName = subscriberName;
    }

    /**
     * 发布者、订阅者之间发生订阅关系时执行的方法
     *
     * @param subscription 订阅内容
     */
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("订阅成功，订阅者: " + subscriberName);
        // Subscriber开始接收数据，1表示接收1次
        subscription.request(1);
        // 为了在后面onNext方法中支持重复订阅
        this.subscription = subscription;
    }

    /**
     * 当有新消息接收时执行的方法
     *
     * @param item 新消息对象
     */
    @Override
    public void onNext(String item) {
        System.out.println("订阅者: " + subscriberName + "-接收数据成功: " + item);
        // 本次消费结束以后，重新订阅，否则无法收到后续消息
        this.subscription.request(1);
    }

    /**
     * 当发生错误时执行的方法
     *
     * @param throwable 异常对象
     */
    @Override
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
    }

    /**
     * 当发布订阅完成以后执行的方法
     */
    @Override
    public void onComplete() {
        System.out.println("当前订阅完成");
    }
}
